package org.heapbd.simulation;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.TimeZone;
import java.util.zip.GZIPOutputStream;

/**
 * This is a data generation thread for use with the Simulator class.
 * 
 * It generates one file of simulated netflow records, using the
 * client and server IP ranges created by the Simulator class, along 
 * with the port distributions.
 * 
 * Note that event timestamps begin at 2014-01-01 00:00:00; the latest timestamp
 * possible is dependent on the Simulator class, which currently calculates it 
 * based on the amount of data to be generated.
 * 
 * @author Levi
 *
 */
public class DataGenerationThread implements Runnable {

	private Map<String, Integer> clientIps;
	private Map<String, Integer> serverIps;
	private Integer[] clientPorts;
	private Integer[] serverPorts;
	private String outputDir;
	private int threadId;
	private long outputRecordCount;
	private SimpleDateFormat df;
	private int maxSecondsAfter;
	private boolean gzipOutput;
	
	public DataGenerationThread(Map<String, Integer> clientIps, Map<String, Integer> serverIps, 
			Integer[] clientPorts, Integer[] serverPorts, String outputDir, 
			int threadId, long outputRecordCount, int maxSecondsAfter, boolean gzipOutput)
	{
		this.clientIps 		= clientIps;
		this.serverIps 		= serverIps;
		this.clientPorts 	= clientPorts;
		this.serverPorts 	= serverPorts;
		this.outputDir 		= outputDir;
		this.threadId 		= threadId;
		this.outputRecordCount = outputRecordCount;
		this.df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		this.df.setTimeZone(TimeZone.getTimeZone("UTC"));
		this.maxSecondsAfter = maxSecondsAfter;
		this.gzipOutput 	= gzipOutput;
	}
	
	public void run() {
		Random random = new Random(1000L * threadId);
		String[] clientIpArray = clientIps.keySet().toArray(new String[clientIps.size()]);
		Arrays.sort(clientIpArray);
		List<String> serverIpList = new ArrayList<String>();
		for (String serverIp : serverIps.keySet())
		{
			int activity = serverIps.get(serverIp);
			for (int x = 0; x < activity; x++)
			{
				serverIpList.add(serverIp);
			}
		}
		String[] serverIpArray = serverIpList.toArray(new String[serverIpList.size()]);
		Arrays.sort(serverIpArray);
		
		try {
			Date eventDateTime = df.parse("2014-01-01 00:00:00");
			long eventEarliestSeconds = eventDateTime.getTime() / 1000;
			
			OutputStream out;
			if (! gzipOutput)
			{
				out = new BufferedOutputStream(new FileOutputStream(
						outputDir + File.separator + "netflow_" + threadId + ".txt"));
			}
			else 
			{
				out = new BufferedOutputStream(new GZIPOutputStream(new FileOutputStream(
						outputDir + File.separator + "netflow_" + threadId + ".txt.gz")));
			}
			
			long writtenRecordCount = 0L;
			
			while (writtenRecordCount < outputRecordCount)
			{
				for (int x = 0; x < clientIpArray.length 
						&& writtenRecordCount < outputRecordCount; x++)
				{
					String clientIp = clientIpArray[x];
					int activityLevel = clientIps.get(clientIp);
					
					// use the activity level as the count of network connections
					// to generate for this client IP
					for (int y = 0; y < activityLevel 
							&& writtenRecordCount < outputRecordCount; y++)
					{
						boolean p2p = random.nextInt(100) == 0 ? true : false;
						
						String serverIp = "";
						if (p2p)
						{
							serverIp = clientIpArray[random.nextInt(clientIpArray.length)];
						}
						else
						{
							serverIp = serverIpArray[random.nextInt(serverIpArray.length)];
						}
						int clientPort = clientPorts[random.nextInt(clientPorts.length)];
						
						int serverPort = 0;
						if (p2p)
						{
							serverPort = clientPorts[random.nextInt(clientPorts.length)];
						}
						else
						{
							serverPort = serverPorts[random.nextInt(serverPorts.length)];
						}
						
						long eventSecondsSinceEpoch = eventEarliestSeconds 
								+ random.nextInt(maxSecondsAfter);
						eventDateTime.setTime(eventSecondsSinceEpoch * 1000);
						int duration = random.nextInt(3600);
						int octetCount = random.nextInt(10000);
						if (random.nextInt(1000) == 0)
						{
							octetCount *= 1000;
						}
						
						String outputRecord = df.format(eventDateTime) + "|" 
								+ clientIp + "|" + clientPort + "|" + serverIp + "|"
								+ serverPort + "|" + duration + "|" + octetCount + "\n";
						out.write(outputRecord.getBytes("UTF-8"));
						
						writtenRecordCount++;
					}
				}
			}
			
			out.close();
		} catch (IOException ioe) {
			System.err.println("Failed to complete data generation for thread " + threadId);
			ioe.printStackTrace();
		} catch (ParseException pe) {
			System.err.println("Failed to complete data generation for thread " + threadId);
			pe.printStackTrace();
		}
		
	}

}
